#!/bin/env python
# -*- coding: utf-8 -*-

"""
File Name: request.py
Author: morre <morre@vip.qq.com>
Create Time: 2022/12/31 17:02:02
Brief:
"""
from __future__ import absolute_import
import logging
from collections import namedtuple

from hbase_client.pb.Client_pb2 import GetRequest, MutateRequest, ScanRequest, Column, MutationProto, MultiRequest
from hbase_client.pb.HBase_pb2 import TimeRange
from .filters import _to_filter, PrefixFilter
from .hbase_exceptions import MalformedFamilies, MalformedValues

Action = namedtuple("Action", ["method", "table", "key", "values", "timestamp"])


class Request(object):
    """请求包体构造类"""
    def __init__(self, req_type, pb):
        self.type = req_type
        self.pb = pb
    #  Converts a dictionary specifying ColumnFamilys -> Qualifiers
    #  into the Column pb type.
    #
    #    Families should look like
    #    {
    #        "columnFamily1": [
    #            "qual1",
    #            "qual2"
    #        ],
    #        "columnFamily2": [
    #            "qual3"
    #        ]
    #    }
    #  Also support single qualifier
    #    {
    #        "columnFamily1": "qual",
    #        "columnFamily2": [
    #            "qual3"
    #        ]
    #    }

    @staticmethod
    def families_to_columns(fam):
        try:
            cols = []
            for key in fam.keys():
                c = Column()
                c.family = key
                val = fam[key]
                if isinstance(val, list):
                    c.qualifier.extend(val)
                else:
                    c.qualifier.append(val)
                cols.append(c)
            return cols
        except Exception as exc:
            logging.exception("families_to_columns except: %s", str(exc))
            raise MalformedFamilies(
                "families coding must be utf-8 and should look like "
                "{\"columnFamily1\":[\"col1\",\"col2\"], "
                "\"colFamily2\": \"col3\"}") from exc

    # Converts a dictionary specifying ColumnFamilys -> Qualifiers -> Values
    # into the protobuf type.
    #
    #   {
    #      "cf1": {
    #           "mycol": "hodor",
    #           "mycol2": "alsohodor"
    #      },
    #      "cf2": {
    #           "mycolumn7": 24
    #      }
    #   }
    @staticmethod
    def values_to_column_values(val, delete=False):
        try:
            col_vals = []
            for cf in val.keys():
                cv = MutationProto.ColumnValue()
                cv.family = cf
                qual_vals = []
                for qual in val[cf].keys():
                    qv = MutationProto.ColumnValue.QualifierValue()
                    qv.qualifier = qual
                    qv.value = val[cf][qual]
                    if delete:
                        qv.delete_type = 1
                    qual_vals.append(qv)
                cv.qualifier_value.extend(qual_vals)
                col_vals.append(cv)
            return col_vals
        except Exception as exc:
            logging.exception("values_to_column_values except: %s", str(exc))
            raise MalformedValues(
                "values coding must be utf-8 and format should look like"
                "{\"cf1\": {\"mycol\": \"hodor\", \"mycol2\": \"alsohodor\"},"
                "\"cf2\": {\"mycolumn7\": 24}}") from exc

    @staticmethod
    def meta_region(table, key, reverse=True):
        # Table + Family used when requesting meta information from the
        # MetaRegionServer
        #metaTableName = "hbase:meta,,1"
        #metaInfoFamily = {"info": []}
        rq = ScanRequest()
        rq.region.type = 1
        rq.region.value = b"hbase:meta,,1"
        rq.number_of_rows = 1
        rq.close_scanner = True
        rq.scan.column.extend(Request.families_to_columns({b"info": []}))
        if not key:
            key = b""
        if reverse:
            rq.scan.start_row = table + b"\xFF,"
        else:
            rq.scan.start_row = table + b"," + key + b",:"
        rq.scan.reversed = True
        rq.limit_of_rows = 1
        rq.scan.filter.CopyFrom(_to_filter(PrefixFilter(table + b",")))
        return Request("Scan", rq)

    @staticmethod
    def get(region, key, families, filters, exists, time_range):
        pb_filter = _to_filter(filters)
        rq = GetRequest()
        rq.get.row = key
        rq.get.column.extend(Request.families_to_columns(families))
        rq.region.type = 1
        rq.region.value = region.region_name
        if pb_filter is not None:
            rq.get.filter.CopyFrom(pb_filter)
        if time_range:
            time_range_obj = TimeRange()
            setattr(time_range_obj, "from", time_range[0])
            time_range_obj.to = time_range[1]
            rq.get.time_range.CopyFrom(time_range_obj)

        if exists:
            rq.get.existence_only = True
        return Request("Get", rq)

    @staticmethod
    def put(region, key, values, timestamp):
        rq = MutateRequest()
        rq.region.type = 1
        rq.region.value = region.region_name
        rq.mutation.row = key
        rq.mutation.mutate_type = 2
        if timestamp is not None:
            rq.mutation.timestamp = timestamp
        rq.mutation.column_value.extend(Request.values_to_column_values(values))
        return Request("Mutate", rq)

    @staticmethod
    def delete(region, key, values, timestamp):
        rq = MutateRequest()
        rq.region.type = 1
        rq.region.value = region.region_name
        rq.mutation.row = key
        rq.mutation.mutate_type = 3
        if timestamp is not None:
            rq.mutation.timestamp = timestamp
        rq.mutation.column_value.extend(
            Request.values_to_column_values(values, delete=True))
        return Request("Mutate", rq)

    @staticmethod
    def append(region, key, values, timestamp):
        rq = MutateRequest()
        rq.region.type = 1
        rq.region.value = region.region_name
        rq.mutation.row = key
        rq.mutation.mutate_type = 0
        if timestamp is not None:
            rq.mutation.timestamp = timestamp
        rq.mutation.column_value.extend(Request.values_to_column_values(values))
        return Request("Mutate", rq)

    @staticmethod
    def increment(region, key, values, timestamp):
        rq = MutateRequest()
        rq.region.type = 1
        rq.region.value = region.region_name
        rq.mutation.row = key
        rq.mutation.mutate_type = 1
        if timestamp is not None:
            rq.mutation.timestamp = timestamp
        rq.mutation.column_value.extend(Request.values_to_column_values(values))
        return Request("Mutate", rq)

    @staticmethod
    def scan(
        region, batch_size, caching, number_of_rows, start_key, stop_key,
        families,filters, close, scanner_id, reverse, limit_of_rows,
        time_range):
        rq = ScanRequest()
        rq.region.type = 1
        rq.region.value = region.region_name
        rq.number_of_rows = 128
        if number_of_rows:
            rq.number_of_rows = number_of_rows
        if close:
            rq.close_scanner = close
        if scanner_id is not None:
            rq.scanner_id = int(scanner_id)
            return Request("Scan", rq)
        rq.scan.column.extend(Request.families_to_columns(families))
        rq.scan.start_row = start_key
        # 如果stop_row为None会触发proto_message非法错误.
        rq.scan.stop_row = b""
        if stop_key is not None:
            rq.scan.stop_row = stop_key
        if filters is not None:
            rq.scan.filter.CopyFrom(filters)
        if batch_size is not None:
            rq.scan.batch_size = batch_size
        if caching is not None:
            rq.scan.caching = caching
        rq.scan.reversed = reverse
        if limit_of_rows:
            rq.limit_of_rows = limit_of_rows

        if time_range:
            time_range_obj = TimeRange()
            setattr(time_range_obj, "from", time_range[0])
            time_range_obj.to = time_range[1]
            rq.scan.time_range.CopyFrom(time_range_obj)

        return Request("Scan", rq)

    @staticmethod
    # TODO: 临时兼容mget的exists，后续改造request格式
    def multi(region, actions, exists=False):
        _ = exists
        rq = MultiRequest()
        mutate_type_map = {
            "put": 2,
            "delete": 3,
            "append": 0,
            "increment": 1
        }
        ra = rq.regionAction.add()
        ra.region.type = 1
        ra.region.value = region.region_name
        for idx, action in enumerate(actions):
            raa = ra.action.add()
            raa.index = idx
            raa.mutation.row = action.key
            raa.mutation.mutate_type = mutate_type_map[action.method]
            if action.timestamp is not None:
                raa.mutation.timestamp = action.timestamp
            is_delete = False
            if action.method == "delete":
                is_delete = True
            raa.mutation.column_value.extend(
                Request.values_to_column_values(
                    action.values, delete=is_delete))
        return Request("Multi", rq)

    @staticmethod
    def multi_get(region, actions, exists, families, filters):
        rq = MultiRequest()
        ra = rq.regionAction.add()
        ra.region.type = 1
        ra.region.value = region.region_name
        pb_filter = None
        if filters:
            pb_filter = _to_filter(filters)

        pb_family = None
        if families:
            pb_family = Request.families_to_columns(families)

        for index, action in enumerate(actions):
            raa = ra.action.add()
            raa.index = index
            raa.get.row = action.key

            if pb_family:
                raa.get.column.extend(pb_family)
            if pb_filter:
                raa.get.filter.CopyFrom(pb_filter)
            if exists:
                raa.get.existence_only = True

        return Request("Multi", rq)
